package yunjiao.javatutorials.apache.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * 高级生产者实现
 *
 * @author yangyunjiao
 */
@Slf4j
public class KafkaBestPracticeProducer {
    private final KafkaProducer<String, String> producer;
    private final String topic;

    public KafkaBestPracticeProducer(String topic) {
        Properties props = Configs.getProducerConfigs();
        this.producer = new KafkaProducer<>(props);
        this.topic = topic;
        log.info("Kafka生产者初始化完成，主题: {}", topic);
    }

    /**
     * 发送消息 - 异步方式（推荐用于高吞吐场景）
     */
    public void sendMessageAsync(String key, String value) {
        // 创建消息头（用于传递元数据）
        List<Header> headers = new ArrayList<>();
        headers.add(new RecordHeader("source", "best-practice-producer".getBytes()));
        headers.add(new RecordHeader("timestamp", String.valueOf(System.currentTimeMillis()).getBytes()));

        // 构建ProducerRecord
        ProducerRecord<String, String> record = new ProducerRecord<>(
                topic,
                null, // 分区，设为null让Kafka根据key自动分配
                System.currentTimeMillis(), // 时间戳
                key,
                value,
                headers
        );

        // 异步发送
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    log.error("消息发送失败: key={}, error={}", key, exception.getMessage());
                    // 这里可以添加重试逻辑或告警
                } else {
                    log.debug("消息发送成功: topic={}, partition={}, offset={}, key={}",
                            metadata.topic(), metadata.partition(), metadata.offset(), key);
                }
            }
        });
    }

    /**
     * 发送消息 - 同步方式（推荐用于需要确认的重要消息）
     */
    public boolean sendMessageSync(String key, String value) {
        try {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
            Future<RecordMetadata> future = producer.send(record);

            // 同步等待响应，设置超时时间
            RecordMetadata metadata = future.get(30, TimeUnit.SECONDS);

            log.info("同步发送成功: partition={}, offset={}, key={}",
                    metadata.partition(), metadata.offset(), key);
            return true;

        } catch (Exception e) {
            log.error("同步发送失败: key={}, error={}", key, e.getMessage());
            return false;
        }
    }

    /**
     * 批量发送消息
     */
    public void sendBatchMessages(List<String> keys, List<String> values) {
        if (keys.size() != values.size()) {
            throw new IllegalArgumentException("keys和values数量必须相同");
        }

        for (int i = 0; i < keys.size(); i++) {
            sendMessageAsync(keys.get(i), values.get(i));
        }

        // 确保所有消息都发送完成
        producer.flush();
        log.info("批量发送完成，共{}条消息", keys.size());
    }

    /**
     * 优雅关闭
     */
    public void close() {
        if (producer != null) {
            try {
                producer.flush(); // 确保所有缓冲消息都发送出去
                producer.close(Duration.ofSeconds(30)); // 优雅关闭，等待最多30秒
                log.info("Kafka生产者已关闭");
            } catch (Exception e) {
                log.error("关闭生产者时发生错误: {}", e.getMessage());
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 使用示例
        KafkaBestPracticeProducer producer = new KafkaBestPracticeProducer("best-practices-topic");

        try {
            // 发送异步消息
            for (int i = 0; i < 10; i++) {
                String key = "key-" + i;
                String value = "{\"id\": " + i + ", \"message\": \"Hello Kafka Best Practices\", \"timestamp\": " + System.currentTimeMillis() + "}";
                producer.sendMessageAsync(key, value);

                // 控制发送速率，避免过快
                Thread.sleep(100);
            }

            // 发送同步消息（重要消息）
            boolean success = producer.sendMessageSync("important-key", "重要业务数据");
            if (success) {
                System.out.println("重要消息发送成功");
            }

        } finally {
            producer.close();
        }
    }
}
